 1.CkickHouse副本和分片之副本
   
   ReplicatedMergeTree
   zk: 实现多个实例之间的通信。
   1).副本的特点
   作为数据副本的主要实现载体，ReplicatedMergeTree在设计上有一些显著特点:
       依赖ZooKeeper：在执行INSERT和ALTER查询的时候，ReplicatedMergeTree需要借助ZooKeeper的分布式协
同能力，以实现多个副本之间的同步。但是在查询副本的时候，并不需要使用ZooKeeper。关于这方面的更多
信息，会在稍后详细介绍。
       表级别的副本：副本是在表级别定义的，所以每张表的副本配置都可以按照它的实际需求进行个性化定义，包
括副本的数量，以及副本在集群内的分布位置等。
       多主架构（Multi Master）：可以在任意一个副本上执行INSERT和ALTER查询，它们的效果是相同的。这些操
作会借助ZooKeeper的协同能力被分发至每个副本以本地形式执行。
       Block数据块：在执行INSERT命令写入数据时，会依据max_insert_block_size的大小（默认1048576行）将数
据切分成若干个Block数据块。所以Block数据块是数据写入的基本单元，并且具有写入的原子性和唯一性。
       原子性：在数据写入时，一个Block块内的数据要么全部写入成功，要么全部失败。
	   唯一性：在写入一个Block数据块的时候，会按照当前Block数据块的数据顺序、数据行和数据大小等指标，计
算Hash信息摘要并记录在案。在此之后，如果某个待写入的Block数据块与先前已被写入的Block数据块拥有相同的Hash
摘要（Block数据块内数据顺序、数据大小和数据行均相同），则该Block数据块会被忽略。这项设计可以预防由异常
原因引起的Block数据块重复写入的问题。如果只是单纯地看这些特点的说明，可能不够直观。没关系，接下来会逐步展
开，并附带一系列具体的示例。
   2).zk的配置方式
   新建配置文件 /etc/clickhouse-server/config.d/metrika.xml
<yandex>
	<zookeeper-servers>
		<node index="1">
			<host>linux121</host>
			<port>2181</port>
		</node>
		<node index="2">
			<host>linux122</host>
			<port>2181</port>
		</node>
		<node index="3">
			<host>linux123</host>
			<port>2181</port>
		</node>
	</zookeeper-servers>
</yandex>
   修改配置文件/etc/clickhouse-server/config.xml
<include_from>/etc/clickhouse-server/config.d/metrika.xml</include_from>
<zookeeper incl="zookeeper-servers" optional="true" />
   3).副本的定义形式
REATE TABLE table_name
(
EventDate DateTime,
CounterID UInt32,
UserID UInt32
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/table_name', '{replica}')
PARTITION BY toYYYYMM(EventDate)
ORDER BY (CounterID, EventDate, intHash32(UserID))
SAMPLE BY intHash32(UserID)

   创建一个副本表
linux121 :) create table replicated_sales_5( id String, price Float64, create_time DateTime
)ENGINE=ReplicatedMergeTree('/clickhouse/tables/01/replicated_sales_5','ch5.nauu.com')
PARTITION BY toYYYYMM(create_time) ORDER BY id;

CREATE TABLE replicated_sales_5
(
`id` String,
`price` Float64,
`create_time` DateTime
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/01/replicated_sales_5', 'linux121')
PARTITION BY toYYYYMM(create_time)
ORDER BY id

   ENGINE = ReplicatedMergeTree('/clickhouse/tables/01/replicated_sales_5', 'linux121')说明：
   /clickhouse/tables: 约定俗称的路径固定前缀
   01:分片编号
   replicated_sales_5：数据表的名字，建议与物理表名相同
   linux121:在zk中创建副本的名称，约定俗成使用服务器的名称。
   
   CH提供了一张zookeeper的代理表，可用通过SQL语句读取远端ZooKeeper的数据：
linux121 :) select * from zookeeper where path = '/clickhouse';

SELECT *
FROM zookeeper
WHERE path = '/clickhouse';


┌─name─┬─value─┬───────czxid─┬───────mzxid─┬───────────────ctime─┬───────────────mtime─┬─ver
sion
│ 01 │ │ 21474837087 │ 21474837087 │ 2020-08-24 14:52:24 │ 2020-08-24 14:52:24 │
0
└──────┴───────┴─────────────┴─────────────┴─────────────────────┴─────────────────────┴─

   通过zk查看
   WATCHER::
WatchedEvent state:SyncConnected type:None path:null
[zk: localhost:2181(CONNECTED) 0] ls
[zk: localhost:2181(CONNECTED) 1] ls /
[itmayeidu_Lasting, cluster, controller, brokers, zookeeper, admin, isr_change_notification,
log_dir_event_notification, controller_epoch, clickhouse, edu-front-boot,
sentinel_rule_config, consumers, front, latest_producer_id_block, config, hbase]
[zk: localhost:2181(CONNECTED) 2] ls /clickhouse
[tables, task_queue]
[zk: localhost:2181(CONNECTED) 3] ls /clickhouse/tables/
Command failed: java.lang.IllegalArgumentException: Path must not end with / character
[zk: localhost:2181(CONNECTED) 4] ls /clickhouse/tables
[01]
[zk: localhost:2181(CONNECTED) 5] ls /clickhouse/tables/01
[replicated_sales_5]
[zk: localhost:2181(CONNECTED) 6] ls /clickhouse/tables/01/replicated_sales_5
[metadata, temp, mutations, log, leader_election, columns, blocks,
nonincrement_block_numbers, replicas, quorum, block_numbers]
[zk: localhost:2181(CONNECTED) 7]

 2.ReplicatedMergeTree原理
   
   1).数据结构
[zk: localhost:2181(CONNECTED) 6] ls /clickhouse/tables/01/replicated_sales_5
[metadata, temp, mutations, log, leader_election, columns, blocks,
nonincrement_block_numbers, replicas, quorum, block_numbers]
   元数据
   metadata:元数信息： 主键、采样表达式、分区键
   columns:列的字段的数据类型、字段名
   replicats:副本的名称
   
   标志：
   leder_election:主副本的选举路径
   blocks:hash值（复制数据重复插入）、partition_id
   max_insert_block_size: 1048576行
   block_numbers:在同一分区下block的顺序
   quorum:副本的数据量
   
   操作类：
   log:log-000000 常规操作
   mutations: delete update
   replicas:
   
   Entry：
   LogEntry和MutationEntry
   
format version: 4
create_time: 2022-05-13 01:08:29
source replica: linux121
block_id: 202008_780388339403124112_11960209063776297222
get
202008_0_0_0

   get:指令（获取数据的指令）
   谁会获取这个指令？ --- - linux122会获取，并执行
   202008_0_0_0：分区信息、告诉linux122你要获取哪一个分区的数据
   2).副本协同的核心流程
   (1).INSERT
   在linux121机器上创建一个副本实例：
create table a1(
id String,
price Float64,
create_time DateTime
)ENGINE=ReplicatedMergeTree('/clickhouse/tables/01/a1','linux121')
PARTITION BY toYYYYMM(create_time)
ORDER BY id;
   
       根据zk_path初始化所有的zk节点
       在replicas节点下注册自己的副本实例linux121
       启动监听任务，监听/log日志节点
       参与副本选举，选出主副本。选举的方式是向leader_election/插入子节点，第一个插入成功的
副本就是主副本  
   
   创建第二个副本实例：
create table a1(
id String,
price Float64,
create_time DateTime
)ENGINE=ReplicatedMergeTree('/clickhouse/tables/01/a1','linux122')
PARTITION BY toYYYYMM(create_time)
ORDER BY id; 

   参与副本选举，linux121副本成为主副本。
   向第一个副本实例插入数据：  
insert into table a1 values('A001',100,'2020-08-20 08:00:00');
   插入命令执行后，在本地完成分区目录的写入，接着向block写入该分区的block_id   
[zk: localhost:2181(CONNECTED) 8] ls /clickhouse/tables/01/a1/blocks
[202008_780388339403124112_11960209063776297222]   
   如果设置了inser_quorum参数，且insert_quorum>=2,则linux122会进一步监控已完成写入操作的副本个数，
直到写入副本个数>= insert_quorum的时候，整个写入操作才算完成。
   接下来，linux121副本发起向log日志推送操作日志[log-0000000000]
[zk: localhost:2181(CONNECTED) 12] ls /clickhouse/tables/01/a1/log
[log-0000000000] 
   操作日式的内容为：  
[zk: localhost:2181(CONNECTED) 13] get /clickhouse/tables/01/a1/log/log-0000000000
format version: 4
create_time: 2022-05-13 01:08:29
source replica: linux121
block_id: 202008_780388339403124112_11960209063776297222
get
202008_0_0_0

cZxid = 0x32000000e4
ctime = Fri May 13 01:08:30 CST 2022
mZxid = 0x32000000e4
mtime = Fri May 13 01:08:30 CST 2022
pZxid = 0x32000000e4
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 150
numChildren = 0
[zk: localhost:2181(CONNECTED) 14] 

   LogEntry:
   source replica: 发送这条Log指令的副本来源，对应replica_name
   ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/table_name', '{replica_name}')
   get: 操作指令类型
        get:从远程副本下载分区
        merge：合并分区
        mutate：MUTATION操作
   block_id：当前分区的blockId,对应/blocks路径下的子节点名称
   202008_0_0_0： 当前分区目录的名称
   
   从日志内容可以看到，操作类型为get下载，需要下载的分区是202008_0_0_0，其余所有副本都会基于Log日志以相
同的顺序执行。
   接下来：第二个副本实例拉取Log日志：
   linux122会一直监听/log节点变化，当linux121推送了/log/log-0000000000 之后,linux122便会触发日志的拉取任务，
并更新log_pointer,
[zk: localhost:2181(CONNECTED) 14] get /clickhouse/tables/01/a1/replicas/linux122/log_pointer
1
cZxid = 0x32000000d5
ctime = Fri May 13 01:07:58 CST 2022
mZxid = 0x32000000e6
mtime = Fri May 13 01:08:30 CST 2022
pZxid = 0x32000000d5
cversion = 0
dataVersion = 2
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 1
numChildren = 0
   
   在拉取LogEntry之后，它并不会立即执行，而是将其转成任务对象放入队列
[zk: localhost:2181(CONNECTED) 15] ls /clickhouse/tables/01/a1/replicas/linux122/queue
[queue-0000000000]

[zk: localhost:2181(CONNECTED) 16] get /clickhouse/tables/01/a1/replicas/linux122/queue/queue-0000000000
format version: 4
create_time: 2022-05-13 01:08:29
source replica: linux121
block_id: 202008_780388339403124112_11960209063776297222
get
202008_0_0_0

cZxid = 0x32000000e6
ctime = Fri May 13 01:08:30 CST 2022
mZxid = 0x32000000e6
mtime = Fri May 13 01:08:30 CST 2022
pZxid = 0x32000000e6
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 150
numChildren = 0

   第二个副本实例向其他副本发起下载请求。
   当看到type为get的时候，ReplicatedMergeTree会明白在远端的其它副本已经成功写入了数据分区，并根据
log_pointer下标做大的下载数据。
   linux121的DataPartsExchange端口服务就收到调用请求,在得知对方来意之后，根据参数做出相应，将本地的
202008_0_0_0基于DataPartsExchange的服务相应发送给linux122

 3.分片
   
   配置文件
<yandex>
	<clickhouse_remote_servers>
		<perftest_3shards_1replicas> --- 集群的名字
			<shard>
				<internal_replication>true</internal_replication>
				<replica>
					<host>linux121</host>
					<port>9000</port>
				</replica>
			</shard>
			<shard>
				<replica>
					<internal_replication>true</internal_replication>
					<host>linux122</host>
					<port>9000</port>
				</replica>
			</shard>
			<shard>
				<internal_replication>true</internal_replication>
				<replica>
					<host>linux123</host>
					<port>9000</port>
				</replica>
			</shard>
		</perftest_3shards_1replicas>
	</clickhouse_remote_servers>

	<zookeeper-servers>
		<node index="1">
			<host>linux121</host>
			<port>2181</port>
		</node>
		<node index="2">
			<host>linux122</host>
			<port>2181</port>
		</node>
		<node index="3">
			<host>linux123</host>
			<port>2181</port>
		</node>
	</zookeeper-servers>
	<macros>
		<shard>01</shard>
		<replica>linux121</replica>
	</macros>
	<networks>
		<ip>::/0</ip>
	</networks>
	<clickhouse_compression>
		<case>
			<min_part_size>10000000000</min_part_size>
			<min_part_size_ratio>0.01</min_part_size_ratio>
			<method>lz4</method>
		</case>
	</clickhouse_compression>
</yandex>

create table clutertable on cluster perftest_3shards_1replicas(id UInt64)
engine = ReplicatedMergeTree('/clickhouse/tables/{shard}/clutertable','{replica}')
order by id;

 4.Distributed用法

   Distributed表引擎：
   all ： 全局查询的
   local：真正的保存数据的表
 
 5.Distributed
   
   分布式引擎，本身不存储数据, 但可以在多个服务器上进行分布式查询。 读是自动并行的。读取时，远程服务器表的
索引（如果有的话）会被使用。
   Distributed(cluster_name, database, table [, sharding_key])
   参数解析：
   cluster_name - 服务器配置文件中的集群名,在/etc/metrika.xml中配置的
   database – 数据库名
   table – 表名
   sharding_key – 数据分片键

create table distributed_all on cluster perftest_3shards_1replicas (id UInt16)
ENGINE=Distributed(perftest_3shards_1replicas,default,distributed_local, rand());

create table distributed_local on cluster perftest_3shards_1replicas (id UInt16)
engine=MergeTree() order by id partition by id;

insert into table distributed_all values(1);
insert into table distributed_all values(2);
insert into table distributed_all values(3);
   案例演示
   
   1).在linux121，linux122,linux123上分别创建一个表distributed_local
   create table t(id UInt16, name String) ENGINE=TinyLog;
   2).在三台机器的t表中插入一些数据
insert into t(id, name) values (1, 'zhangsan');
insert into t(id, name) values (2, 'lisi');   
   3).在linux121上创建分布式表
create table dis_table(id UInt16, name String)
ENGINE=Distributed(perftest_3shards_1replicas, default, t, id);   
   4).往dis_table中插入数据
insert into dis_table select * from t   
   5).查看数据量
:) select count() from dis_table;

FROM dis_table
┌─count()─┐
│ 8 │
└─────────┘

:) select count() from t;

SELECT count()
FROM t
┌─count()─┐
│ 3 │
└─────────┘

   可以看到每个节点大约有1/3 的数据
